package org.example.table_type;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class FlinkBatchFormatTable {
    public static void main(String[] args) throws Exception {
        // 使用批处理模式创建TableEnvironment
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inBatchMode() // 设置为批处理模式
                .build();
        TableEnvironment tableEnv = TableEnvironment.create(settings);

        // 创建Paimon catalog，使用默认文件系统
        String createCatalogSQL = "CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'metastore' = 'hive',\n" +
                "    'uri' = 'thrift://hadoop1:9083'" +
                ");";
        System.out.println("执行SQL: \n" + createCatalogSQL);
        tableEnv.executeSql(createCatalogSQL);

        // 先删除表（如果存在）
        String dropTableSQL = "DROP TABLE IF EXISTS `paimon`.`default`.`my_csv_table`;";
        System.out.println("执行SQL: \n" + dropTableSQL);
        tableEnv.executeSql(dropTableSQL);

        // 创建Paimon原生表
        String createTableSQL = "CREATE TABLE `paimon`.`default`.`my_csv_table` (\n" +
                "    a INT PRIMARY KEY NOT ENFORCED,\n" +
                "    b STRING\n" +
                ") WITH (\n" +
                "    'type'='format-table',\n" +
                "    'file.format'='csv',\n" +
                "    'field-delimiter'=','\n" +
                ");";
        System.out.println("执行SQL: \n" + createTableSQL);
        tableEnv.executeSql(createTableSQL);
        
        // 打印表的存储路径
        // 使用SHOW CREATE TABLE语句获取表的存储路径
        String showCreateTableSQL = "SHOW CREATE TABLE `paimon`.`default`.`my_csv_table`";
        System.out.println("执行SQL: \n" + showCreateTableSQL);
        System.out.println("表创建语句:");
        tableEnv.executeSql(showCreateTableSQL).print();

        // 查询表结构
        String descTableSQL = "DESC `paimon`.`default`.`my_csv_table`";
        System.out.println("执行SQL: \n" + descTableSQL);
        System.out.println("表结构信息:");
        tableEnv.executeSql(descTableSQL).print();
        
        // 插入一条数据到表中
        String insertSQL = "INSERT INTO `paimon`.`default`.`my_csv_table` VALUES\n" +
                "(1, '测试数据')\n" +
                ";";
        System.out.println("执行SQL: \n" + insertSQL);
        tableEnv.executeSql(insertSQL);
        
        // 查询并打印插入的数据
        String querySQL = "SELECT * FROM `paimon`.`default`.`my_csv_table`";
        System.out.println("执行SQL: \n" + querySQL);
        System.out.println("查询结果:");
        tableEnv.executeSql(querySQL).print();
        
        // 确保所有操作都已经完成
        System.out.println("查询执行完毕");
    }
}